package cn.yi.fast.mq.rocketmq.manager;

import cn.yi.fast.mq.rocketmq.listener.DefaultMessageListener;
import cn.yi.fast.mq.rocketmq.message.handler.MessageHandler;
import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.common.UtilAll;
import com.alibaba.rocketmq.common.consumer.ConsumeFromWhere;
import com.alibaba.rocketmq.common.protocol.heartbeat.MessageModel;
import com.google.common.base.Joiner;
import com.google.common.collect.Maps;
import lombok.Setter;

import java.util.Map;
import java.util.Objects;

/**
 * 消息处理Manager
 */
@Setter
public class DefaultMessageConsumerManager {

    private String consumerGroup;

    private String namesrvAddr;

    private MessageModel messageModel;

    private DefaultMQPushConsumer defaultMQPushConsumer;

    private String consumeTimestamp = minutesBeforeHumanStr(30);

    /**
     * 拉取轮询间隔时间
     */
    private int pullInterval = 0;

    /**
     * 向broker消息拉取一次的数量
     */
    private int pullBatchSize = 32;

    /**
     * 并发消费时，一次消费消息的数量
     */
    private int consumeMessageBatchMaxSize = 1;

    /**
     * 线程池自动调整
     */
    private int consumeThreadMin = 20;

    /**
     * 线程池自动调整
     */
    private int consumeThreadMax = 64;


    private DefaultMessageListener defaultMessageListener;


    public void init() throws MQClientException {
        defaultMQPushConsumer = new DefaultMQPushConsumer(consumerGroup);
        defaultMQPushConsumer.setNamesrvAddr(namesrvAddr);
        defaultMQPushConsumer.setMessageModel(messageModel);
        defaultMQPushConsumer.setConsumeThreadMax(consumeThreadMax);
        defaultMQPushConsumer.setConsumeThreadMin(consumeThreadMin);
        defaultMQPushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        defaultMQPushConsumer.setConsumeTimestamp(consumeTimestamp);
        defaultMQPushConsumer.setPullBatchSize(pullBatchSize);
        defaultMQPushConsumer.setConsumeMessageBatchMaxSize(consumeMessageBatchMaxSize);
        defaultMQPushConsumer.setPullInterval(pullInterval);

        if(Objects.isNull(defaultMessageListener)){
            throw new RuntimeException("监听器不存在...");
        }
        Map<String, String> subscribeMap = Maps.newHashMap();
        Map<String,Map<String,MessageHandler>> topic2TagHandler = defaultMessageListener.getTopic2TagHandler();
        topic2TagHandler.forEach((topic,tagHandlerMap) ->{
            subscribeMap.put(topic, Joiner.on(" || ").join(tagHandlerMap.keySet()));
        });

        defaultMQPushConsumer.setSubscription(subscribeMap);

        defaultMQPushConsumer.setMessageListener(defaultMessageListener);

        defaultMQPushConsumer.start();

    }

    public void destory(){
        if(Objects.nonNull(defaultMessageListener)){
            defaultMQPushConsumer.shutdown();
        }
    }


    private String minutesBeforeHumanStr(int minutes) {
        if (minutes < 0) {
            throw new RuntimeException("时间不可小于0");
        }
        return UtilAll.timeMillisToHumanString3(System.currentTimeMillis() - 1000 * 60 * minutes);
    }
}
